package com.lg.algorithm.collections.queue.disruptor;

import java.util.UUID;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;


/**
 * 多生产者，多消费者
 *
 * 数组长度2的倍数
 *  位运算运算，加快定位的速度，与hashmap的大小，同一个道理
 *
 * 环形数组
 *  为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好
 *  数组游标是long类型，即使100万QPS的处理速度，也需要30万年才能用完
 *  下标为运算，就可以定位数组位置，位运算自动成环
 *
 * 避免内存伪共享
 *
 * 无锁设计
 *  每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。
 *
 * 生产过量，怎么控制
 *      获取数组下标的时候控制
 * 消费过量，怎么控制
 *      获取数组下标的时候控制
 */
public class DisruptorMain2 {

    public static void main(String[] args) throws InterruptedException {

//        int bufferSize = 1024 * 1024;
        int bufferSize = 4;

        int workThreadNum = 3;
        //1 创建RingBuffer
        /**
         * 所有的ringBuffer.sequenceBarrier
         *   都有cursor，在抽象类中
         *   ProducerType.MULTI 中增加了 availableBuffer ，在实现类 MultiProducerSequencer 中
         */
        RingBuffer<Order> ringBuffer =
                RingBuffer.create(ProducerType.MULTI,
                        new EventFactory<Order>() {
                            @Override
                            public Order newInstance() {
                                return new Order();
                            }
                        },
                        bufferSize ,
                        new BlockingWaitStrategy());
        
        //2 通过ringBuffer 创建一个屏障
        /**
         * 这里面有 角标、阻塞规则等
         */
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        
        //3 创建含有10个消费者的数组:
        Consumer[] consumers = new Consumer[workThreadNum];

//        当消费者 比消费者组线程多的时候会出问题
//        Consumer[] consumers = new Consumer[10];
        for(int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("C" + i);
        }
        
        //4 构建多消费者工作池
        /**
         * workProcessors 中有消费者该有的所有东西，是个数组和消费者数量一样
         */
        WorkerPool<Order> workerPool = new WorkerPool<Order>(
                ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(),
                consumers);
        
        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        
        //6 启动workerPool
        workerPool
        .start(Executors.newFixedThreadPool(workThreadNum));
        
        CyclicBarrier barrier = new CyclicBarrier(workThreadNum);
        for(int i = 0; i < workThreadNum; i++) {
            final Producer producer = new Producer(ringBuffer);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            System.err.println(Thread.currentThread().getName() + "--> barrier.await() start");
                            barrier.await();
                            System.err.println(Thread.currentThread().getName() + "--> barrier.await() end");
                            for(int j = 0; j < workThreadNum; j++) {
                                producer.sendData(UUID.randomUUID().toString());

                                TimeUnit.MINUTES.sleep(5);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
        
        System.err.println("----------线程创建完毕，开始生产数据----------");

        Thread.sleep(10000);
        
        System.err.println("任务总数:" + consumers[2].getCount());
    }
    
    static class EventExceptionHandler implements ExceptionHandler<Order> {

        @Override
        public void handleEventException(Throwable ex, long sequence, Order event) {
        }

        @Override
        public void handleOnStartException(Throwable ex) {
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
        }

    }
    
}